package ex.datastream.libraries;

import ex.datastream.ApiFrame;
import ex.datastream.functions.timestampAssigner.MyAssigner;
import ex.datatypes.TaxiRide;
import ex.util.generator.TaxiRideGeneratorFromMysql;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * 复杂事件处理（Complex Event Processing），用于识别输入流中符合指定规则的事件，并按照指定方式输出。
 * See https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/libs/cep/#individual-patterns
 */
public class CEP01 extends ApiFrame {



    public static void main(String[] args) throws Exception {
        SerializableTimestampAssigner<TaxiRide> s = new MyAssigner();
        CEP01 cep01 = new CEP01();
        cep01.getEnv();
        cep01.env.setParallelism(1);
        DataStream<TaxiRide> rideDataStream= cep01.env.addSource(new TaxiRideGeneratorFromMysql());

        SingleOutputStreamOperator<TaxiRide>outputStreamOperator= rideDataStream
                .assignTimestampsAndWatermarks(
                    WatermarkStrategy
                            .<TaxiRide>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                            .withTimestampAssigner(s)

                );
        KeyedStream<TaxiRide,Long> stringKeyedStream=outputStreamOperator.keyBy(taxiRide ->taxiRide.driverId);
        stringKeyedStream.print();

        test2(stringKeyedStream);
        cep01.env.execute();
    }





    private static void test2( KeyedStream<TaxiRide,Long> stream ) {
        //定义pattern ,寻找连续搜索帽子的用户
        //组合模式
//        严格连续(严格紧邻)
        Pattern<TaxiRide,?> pattern =
                Pattern.<TaxiRide>
                        begin("first")
                        .where(new SimpleCondition<TaxiRide>() {
                            @Override
                            public boolean filter(TaxiRide taxiRide) throws Exception {
                                System.out.println("first"+taxiRide.driverId);
                                return taxiRide.driverId==2013000108;
                            }
                        })
                .next("second")
//                        .followedBy("second")
                        .where(new SimpleCondition<TaxiRide>() {
                            @Override
                            public boolean filter(TaxiRide taxiRide) throws Exception {
                                System.out.println("second"+taxiRide.driverId);
                                return taxiRide.driverId==2013000108;
                            }
                        });
//        System.out.println(JSON.toJSONString(pattern));
        PatternStream patternStream = CEP.pattern(stream, pattern);
        SingleOutputStreamOperator matchStream = patternStream.select(new PatternSelectFunction<TaxiRide, String>() {
            @Override
            public String select(Map<String, List<TaxiRide>> pattern) throws Exception {
                List<TaxiRide> firsts=  pattern.get("first");
                List<TaxiRide> seconds=  pattern.get("second");
                TaxiRide first = pattern.get("first").get(0);
                TaxiRide second = pattern.get("second").get(0);
                String out = String.format("连续2次搜索 \n第1次%s;\n第2次:%s;", first.toString(), second.toString());
//                String out1=JSON.toJSONString(firsts);
//                String out2=JSON.toJSONString(seconds);
//                return out1="\n"+out2;
                return out;
            }
        });
        matchStream.print("warning");
    }
}
